package ioc

import (
	"github.com/IBM/sarama"
	"github.com/spf13/viper"
	"go-admin/server/internal/events"
	"go-admin/server/internal/events/log"
)

// InitKafka 初始化Kafka 配置 返回 sarama.Client 给 Producer Consumers 使用
func InitKafka() sarama.Client {
	type Config struct {
		Addr []string `yaml:"addr"`
	}
	saramaCfg := sarama.NewConfig()
	saramaCfg.Producer.Return.Successes = true          // 设置 Producer 在成功发送消息后，是否将消息的 partition 和 offset 返回给客户端
	saramaCfg.Producer.RequiredAcks = sarama.WaitForAll // 设置 Producer 等待所有 ISR（In-Sync Replicas）确认的策略
	saramaCfg.Producer.Retry.Max = 3                    // 设置 Producer 在消息发送失败时的最大重试次数
	var cfg Config
	err := viper.UnmarshalKey("kafka", &cfg)
	if err != nil {
		panic(err)
	}
	client, err := sarama.NewClient(cfg.Addr, saramaCfg)
	if err != nil {
		panic(err)
	}
	return client
}

// NewSyncProducer 创建同步发送消息客户端连接
func NewSyncProducer(client sarama.Client) sarama.SyncProducer {
	res, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		panic(err)
	}
	return res
}

// NewConsumers 消费者 把全部的消费者存放到 events.Consumer 用于启动时开启消费者
func NewConsumers(c1 *log.InteractiveReadEventGormLogConsumer) []events.Consumer {
	return []events.Consumer{c1}
}
